package com.controller;


import com.req.SiteDeviceInfo;
import com.req.StatusDataReq;
import org.apache.commons.lang.StringUtils;
import org.springframework.util.ObjectUtils;
import redis.clients.jedis.JedisPool;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.init.InitConfig;
import com.req.TempDataReq;
import com.util.CsvUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import redis.clients.jedis.Jedis;

import javax.validation.Valid;
import java.lang.reflect.Type;
import java.util.Date;
import java.util.List;

@RestController
@RequestMapping("/receive")
public class QueryDataController {
    final static Logger logger = LoggerFactory.getLogger(QueryDataController.class);

    //kafka发送消息的topic
    final static String topic = CsvUtil.getSysProperty("kafka.topic");

    //kafka发送告警消息的topic
    final static String topic_alarm = CsvUtil.getSysProperty("kafka.topic_alarm");

    //kafka发送站点状态消息的topic
    final static String topic_status = CsvUtil.getSysProperty("kafka.topic_status");

    //kafka发送站点设备信息的topic
    final static String topic_info = CsvUtil.getSysProperty("kafka.topic_site_info");

    //维护站点在redis中的心跳 key
    final static String beat = "device:heartbeat%s";

    @Autowired
    JedisPool redisPool;

    //实时温度数据
    //目前修改为多条  [[{站点编号},{采集时间},{点1数据},{点2数据},{点3数据},..{点n数据}],[{站点编号},{采集时间},{点1数据},{点2数据},{点3数据},..{点n数据}]]
    //data 数据格式: [{站点编号},{采集时间},{点1数据},{点2数据},{点3数据},..{点n数据}]
    @PostMapping("/tempData")
    public String sendPeriodData(@RequestBody @Valid TempDataReq req) {
        logger.info("接收到数据"+req.getData());

        try
        {
            Gson gson = new Gson();
            //将将消息发送到kafka
            Type type = new TypeToken<List<List<String>>>() {}.getType();
            List<List<String>> dataList = gson.fromJson(req.getData(), type);
            List<String> tempList = dataList.get(0);
            //该模式会创建大量生产者实例，不使用  主题 键是站点
            //new SendDataToKafkaProducer().produceOne(dataList.get(1),  dataList.get(0), req.getData(),topic);

            InitConfig.simpleKafkaProducer.produceOne(tempList.get(1),  tempList.get(0), req.getData(),topic);

            siteHeartBeat(tempList.get(0));

            return "success";
        }
        catch (Exception e) {
            e.printStackTrace();
            return "fail";
        }

    }



    /**
     * 实时告警数据
     * data 数据格式: {站点编号,采集时间,[通道编号,温度数据,通道编号,温度数据,.....]}
     * //
     */
    @PostMapping("/alarmData")
    public String sendAlarmData(@RequestBody @Valid TempDataReq req) {
        try {
            Gson gson = new Gson();
            //将将消息发送到kafka
            Type type = new TypeToken<List<Object>>() {}.getType();
            List<Object> dataList = gson.fromJson(req.getData(), type);
            //该模式会创建大量生产者实例，不使用
			//new SendDataToKafkaProducer().produceOne(dataList.get(1),  dataList.get(0), req.getData(),topic);
            InitConfig.simpleKafkaProducer.produceOne((String) dataList.get(1),  (String)dataList.get(0), req.getData(),topic_alarm);
            //将站点本次心跳时间更新到redis
            siteHeartBeat((String)dataList.get(0));

            return "success";
        } catch (Exception e) {
            e.printStackTrace();
            return "fail";
        }
    }


    /**
     * 实时站点网络状态数据
     * @param req
     * @return
     */
    @PostMapping("/statusData")
    public String sendStatusData(@RequestBody @Valid StatusDataReq req){
        try{
            Gson gson = new Gson();
            //发送消息到kafka，这个再后面程序中没看到用途，kafka
            InitConfig.simpleKafkaProducer.produceOne(null, req.getSiteNo(), gson.toJson(req),topic_status);
            //将站点本次心跳时间更新到redis
            siteHeartBeat(req.getSiteNo());
            return "success";
        }catch (Exception e){
            return "fail";
        }
    }


    @PostMapping("siteInfo")
    public String siteDeviceInfo(@RequestBody String info){

        logger.info("接收站点设备信息:"+info);

        if(StringUtils.isEmpty(info)){
            return "fail";
        }

        //第一行为站点编号
        // \t是补全当前字符串长度到8的整数倍，最少1个最多8个空格
        // 比如当前字符串长度10，那么\t后长度是16，也就是补6个空格
        //如果当前字符串长度12，此时\t后长度是16，补4个空格
        String[] lines = info.replaceAll(" ", "").replaceAll("\t", "").split("\n");

        if(ObjectUtils.isEmpty(lines)||lines.length<1){
            return "fail";
        }

        String siteNo = lines[0];
        String sn = findInfoValue(lines, "sn");
        String pro = findInfoValue(lines, "pro");
        String softwareVersion = findInfoValue(lines,"softwareVersion");
        String hardwareVersion = findInfoValue(lines,"hardwareVersion");

        SiteDeviceInfo siteDeviceInfo = new SiteDeviceInfo();
        siteDeviceInfo.setSiteNo(siteNo.replaceAll("\r",""));
        siteDeviceInfo.setPro(pro);
        siteDeviceInfo.setSn(sn);
        siteDeviceInfo.setSoftwareVersion(softwareVersion);
        siteDeviceInfo.setHardwareVersion(hardwareVersion);

        Gson gson = new Gson();
        //发送消息到kafka
        InitConfig.simpleKafkaProducer.produceOne(gson.toJson(siteDeviceInfo),topic_info);

        return "success";
    }



    /**
     * @Desc: 维护站点的心跳
     **/
    public void siteHeartBeat(String siteNo){
        //将站点本次心跳时间更新到redis
        String key = String.format(beat,siteNo);
        Jedis jedis = redisPool.getResource();

        logger.info("Jedis_zy_:key="+key+"    ####    valeu="+String.valueOf(new Date().getTime()));
        jedis.set(key,String.valueOf(new Date().getTime()));
        jedis.close(); //记得释放
    }




    public static String findInfoValue(String[] lines,String key){
        if(StringUtils.isEmpty(key)){
            return "";
        }
        key = key.toLowerCase();//转成小写
        for (String line : lines) {
            //public boolean startsWith(String prefix, int toffset)
            //public boolean startsWith(String prefix)
            //prefix -- 前缀。
            //toffset -- 字符串中开始查找的位置
            if(line.toLowerCase().startsWith(key+":")){
                String[] keyWord = line.split(":");
                if(!ObjectUtils.isEmpty(keyWord) && keyWord.length==2){
                    //\r回车  \n换行 \t补全8位
                    return keyWord[1].replaceAll("\r","").replaceAll("\n","").replaceAll("\t","");
                }
            }
        }
        return "";
    }
}
